package com.baishancloud.log.live.sanma

import cn.hutool.json.JSONUtil
import com.baishancloud.log.common.env.StreamEnv
import com.baishancloud.log.common.sink.SinkUtil
import com.baishancloud.log.common.source.SourceUtil
import com.baishancloud.log.format.scala.SanmaOrigin
import org.apache.flink.api.java.utils.ParameterTool
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.slf4j.{Logger, LoggerFactory}

import java.io.Serializable

class StreamingSanmaLive extends Serializable

object StreamingSanmaLive extends Serializable {

  val LOGGER: Logger = LoggerFactory.getLogger(classOf[StreamingSanmaLive])


  def main(args: Array[String]): Unit = {
    val parameterTool = ParameterTool.fromArgs(args)
    println("输入参数为：" + parameterTool.toMap.toString)
    StreamEnv.builder(parameterTool).build()
    val source1: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, ".金华").uid("dd683fb6-3b42-4969-942a-b7ac8dffcf71")
    val source2: DataStream[String] = SourceUtil.kafkaOnlyValue(parameterTool, ".喜鹊").uid("a729ff3d-4ddc-4612-b121-b871dbc845ac")
    source1.union(source2)
      .map(SanmaOrigin.parse(_).orNull).name("SanmaOrigin").uid("4dc8336e-7c9a-4ec5-8eb7-43e536a8155d")
      .filter(_ != null).name("!null").uid("9a6d424b-ed3a-410f-893c-cfd9a51e93c3")
      .filter(new DataFilter(parameterTool)).name("someFilter").uid("4e4227ca-80e5-4cbf-8e04-f5409804486b")
      .process(new SanmaLogRecordParse(parameterTool)).name("LogRecordParse").uid("fde78e8f-9572-42e5-a3d2-60eb2d0e9f2d")
      .keyBy(_.getKey)
      .window(TumblingProcessingTimeWindows.of(Time.minutes(parameterTool.getLong(windowSize, 1))))
      .reduce((x, y) => SanmaLog(x.time_at_1min, x.domain, x.country, x.province, x.isp, x.log_type, x.uv + y.uv)).name("reduce").uid("7e67b2a3-261c-4b88-aec4-77fe782b9f67")
      .map(JSONUtil.toJsonStr(_)).name("toJsonStr").uid("961122a5-9762-4ff4-925b-2bb9a0da4d88")
      .addSink(SinkUtil.starRocksJsonString(parameterTool)).setParallelism(parameterTool.getInt(sinkParallel, 1)).name("starRocks").uid("58ac2db1-2064-44ce-b8ef-531bd003bbe0")
    StreamEnv.execute(parameterTool)
  }

}